#include "TcpConnection.h"
#include "Buffer.h"
// #include "Log/Logging.h"
#include "Channel.h"
#include "EventLoop.h"
#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <functional>
#include <string.h>
#include "Socket.h"
#include <sstream>
#include <thread> 
#include "util.h"

using namespace std::placeholders;

TcpConnection::TcpConnection(EventLoop* loop, const std::string& name, int sockfd)
    : loop_(loop),
    name_(name),
    socket_(new Socket(sockfd)),
    channel_(new Channel(loop, sockfd)),
    state_(kConnecting)
{
    localIpPort_ = toIpPort(getLocalAddr(sockfd));
    peerIpPort_ = toIpPort(getPeerAddr(sockfd));
    channel_->setReadCallback( std::bind(&TcpConnection::handleRead, this) );
    channel_->setWriteCallback( std::bind(&TcpConnection::handleWrite, this) );
    // channel_->setCloseCallback( std::bind(&TcpConnection::connectDestroyed, this) );
    channel_->setErrorCallback( std::bind(&TcpConnection::handleError, this) );
    // loop_->runAfter(5, std::bind(&TcpConnection::handleClose, this));
    setTcpNoDelay(sockfd, 1);
}

TcpConnection::~TcpConnection() {
}

int TcpConnection::fd() { return socket_->fd(); }

void TcpConnection::send(const void* data, int len)
{
    std::string tmp;
    const char* pt = static_cast<const char*>(data);
    for (int i = 0; i < len; ++i) {
        tmp.push_back(*pt);
        pt++;
    }
    send(tmp);
}

void TcpConnection::send(const std::string& message)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        {
            sendStrInLoop(message);
        }
        else
        {
            loop_->runInLoop(std::bind(&TcpConnection::sendStrInLoop, this, message));
        }
    }
    
}

void TcpConnection::send(Buffer* buf)
{
    if (state_ == kConnected)
    {
        if (loop_->isInLoopThread())
        {
            sendInLoop(buf->peek(), buf->readableBytes());
            buf->retrieveAll();
        }
        else
        {
            loop_->runInLoop(std::bind(&TcpConnection::sendStrInLoop, this, buf->retrieveAllAsString()));
        }
    }
}

void TcpConnection::sendStrInLoop(const std::string& message)
{
    sendInLoop(message.data(), message.size());
}

void TcpConnection::sendInLoop(const char* data, size_t len)
{
    ssize_t nwrote = 0;
    ssize_t hasWrite = 0;
    size_t remaining = len;
    bool faultError = false;
    if (state_ == kDisconnected)
    {
        std::cout << "disconnected, give up writing" << std::endl;
        return;
    }
    // if no thing in output queue, try writing directly
    if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
    {
        while (1) {
            nwrote = ::write(channel_->fd(), data+hasWrite, remaining);
            if (nwrote > 0)
            {
                remaining = remaining - nwrote;
                hasWrite += nwrote;
            }
            else if (nwrote == 0) {
                break;
            }
            else // nwrote < 0
            {
                nwrote = 0;
                if (errno != EWOULDBLOCK)
                {
                    // LOG_SYSERR << "TcpConnection::sendInLoop";
                    // printf("TID: %d sendInLoop errno: %2d\t%s\n", getTid(), errno, strerror(errno));
                    if (errno == EPIPE || errno == ECONNRESET)
                    {
                        faultError = true;
                    }
                }
                break;
            }
        }
    }
    
    assert(remaining <= len);
    if (!faultError && remaining > 0)
    {
        size_t oldLen = outputBuffer_.readableBytes();
        outputBuffer_.append(static_cast<const char*>(data)+hasWrite, remaining);
        if (!channel_->isWriting())
        {
            channel_->enableWriting();
        }
    }
}

void TcpConnection::shutdown()
{
    if (state_ == kConnected)
    {
        setState(kDisconnecting);
        loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
    
}

void TcpConnection::shutdownInLoop()
{
    loop_->assertInLoopThread();
    if (!channel_->isWriting())
    {
        socket_->shutdownWrite();
    }
}

void TcpConnection::forceClose()
{
  if (state_ == kConnected || state_ == kDisconnecting)
  {
    setState(kDisconnecting);
    loop_->queueInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
  }
}

void TcpConnection::forceCloseInLoop()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected || state_ == kDisconnecting)
    {
        handleClose();
    }
}

void TcpConnection::startReading() { channel_->enableReading(); }

// 建立连接
void TcpConnection::connectEstablished()
{
    loop_->assertInLoopThread();
    assert(state_ == kConnecting);
    setState(kConnected);
    channel_->enableReading();  // 注册读事件 添加到epoll_wait
    connectionCallback_(shared_from_this());
}

// 销毁连接 该函数结束后TcpConnection析构
void TcpConnection::connectDestroyed()
{
    loop_->assertInLoopThread();
    if (state_ == kConnected)
    {
        setState(kDisconnected);
        channel_->disableAll();
        connectionCallback_(shared_from_this());
    }
    loop_->removeChannel(channel_.get());
}

void TcpConnection::handleRead()
{
    int savedErrno = 0;
    ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
    if ( n < 0 ) {
        errno = savedErrno;
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // 已经读到没有数据
            messageCallback_(shared_from_this(), &inputBuffer_);
        }
        else {
            handleError();
        }
    }
    else {
        // n == 0
        handleClose();
    }
}

void TcpConnection::handleWrite()
{
    loop_->assertInLoopThread();
    if (channel_->isWriting())
    {
        ssize_t n = ::write(channel_->fd(),
                                outputBuffer_.peek(),
                                outputBuffer_.readableBytes());
        if (n > 0)
        {
            outputBuffer_.retrieve(n);
            if (outputBuffer_.readableBytes() == 0)
            {
                channel_->disableWriting();
                if (state_ == kDisconnecting)
                {
                    shutdownInLoop();
                }
            }
            
        }
        else
        {
            if (state_ == kDisconnecting)
            {
              shutdownInLoop();
            }
        }
    }
    else
    {
        std::cout << "Connection is down, no more writing" << std::endl;
    }
}

void TcpConnection::handleClose()
{
    loop_->assertInLoopThread();
    setState(kDisconnected);
    channel_->disableAll();
    connectionCallback_(shared_from_this());
    closeCallback_(shared_from_this());
}

void TcpConnection::handleError()
{
    printf("TID: %d TcpConnection::errno: %2d\t%s\n", getTid(), errno, strerror(errno));
}

bool TcpConnection::isReading() {
    return channel_->isReading();
}